package com.example.spider.cluster.msg;

import com.example.spider.cluster.oss.OssDownloader;
import com.example.spider.provider.m3u8.task.M3u8TsTask;
import com.example.spider.task.AbstractSpiderTask;
import com.example.spider.task.SpiderManager;
import com.example.spider.task.SpiderResultHandlerManager;
import com.rabbitmq.client.AMQP;
import org.shoulder.autoconfigure.condition.ConditionalOnCluster;
import org.shoulder.batch.progress.BatchProgressCache;
import org.shoulder.core.util.JsonUtils;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.autoconfigure.amqp.RabbitHealthContributorAutoConfiguration;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;

import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Configuration
public class MsgConfiguration {

    @Bean
    @ConditionalOnCluster
    public MsgSender msgSender(StreamBridge streamBridge,
                               @Value("${spider.msg.queue.spider.name}") String spiderTaskQueueName,
                               @Value("${spider.msg.queue.spiderResult.name}") String spiderResultQueueName,
                               @Value("${spider.msg.queue.handlerResult.name}") String handlerResultQueueName) {
        return new MsgSender(streamBridge, spiderTaskQueueName, spiderResultQueueName, handlerResultQueueName);
    }

    /**
     * 延迟任务消费者
     *
     * @author lym
     */
    @Configuration
    @ConditionalOnCluster
    public static class MsgConsumer {

        public static LinkedTransferQueue<AbstractSpiderTask<?>> toSenderSpiderTaskQueue = new LinkedTransferQueue<>();
        @Autowired
        private SpiderManager spiderManager;
        @Autowired
        private SpiderResultHandlerManager resultHandlerManager;
        @Autowired
        private BatchProgressCache progressCache;
        @Autowired
        private OssDownloader ossDownloader;

        //@Bean
        //public Consumer<AbstractSpiderTask<?>> onDelayTask() {
        //    // 如果是爬虫限流延迟任务，则放到爬虫待执行队列
        //    return SpiderTaskQueue::putTask;
        //}Message<String>
        @Bean
        @ConditionalOnCluster
        public Consumer<Message<String>> onSpiderResult() {
            // 这里需要手动应答，因为有可能处理不了
            return mqMsg -> {
                AMQP.Channel channel = mqMsg.getHeaders().get(AmqpHeaders.CHANNEL, AMQP.Channel.class);
                Long deliveryTag = mqMsg.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);

                UnionSpiderMsg msg = JsonUtils.parseObject(mqMsg.getPayload(), UnionSpiderMsg.class);
                AbstractSpiderTask<?> task = msg.getTask();
                String processResult = msg.getSpiderResult();
                if (task instanceof M3u8TsTask) {
                    // 检查幂等
                    if (task.getIdempotentPathOptional().isEmpty()) {
                        // 搬运到本机目录
                        String handleResultStr = processResult;
                        // alreadyDown

                        // oss download
                        ossDownloader.download(handleResultStr, task.getIdempotentFilePath(), true);
                        // webdav and delay copy
                    }
                    // 进度 +1
                    String progressId = ((M3u8TsTask) task).getProgressId();
                    progressCache.getProgress(progressId).finishPart(((M3u8TsTask) task).getTsPart().getPartIndex());
                }
                mqMsg.getHeaders().getReplyChannel();
                //resultHandlerManager.handle();
                //channel.basicNack(deliveryTag, false, false);
                //channel.basicAck(deliveryTag, false);
            };
        }

        //@Bean
        public Consumer<Message<String>> onHandlerResult() {
            // 一般是统一保存
            return Message::getPayload;
        }

        @Bean
        public Consumer<Message<String>> onProgressFinished() {
            // 更新任务状态
            return mqMsg -> {
                System.out.println("ok");
            };
        }

        @Bean
        public Consumer<Message<String>> onSpiderTask() {
            return mqMsg -> {
                UnionSpiderMsg msg = JsonUtils.parseObject(mqMsg.getPayload(), UnionSpiderMsg.class);
                AbstractSpiderTask<?> task = msg.getTask();
                if (task == null) {
                    // 未注册反序列化的类 -> 没有 Handler
                    return;
                }
                try {
                    while (!spiderManager.canAccept()) {
                        TimeUnit.SECONDS.sleep(1);
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 放到爬虫线程池，自动确认 ack（重启宕机丢任务不要紧，重试即可，跑过的都有幂等，另外爬虫任务运行时间往往比较长，不建议ack，耗性能）
                spiderManager.doSpider(task);
            };
        }

    }


    /**
     * @see RabbitHealthContributorAutoConfiguration
     */
//    @Bean
//    @ConditionalOnClass({RabbitTemplate.class})
//    @ConditionalOnBean({RabbitTemplate.class})
//    @ConditionalOnCluster(cluster = false)
//    public MockHealthIndicator rabbitHealthIndicator() {
//        return new MockHealthIndicator(Health.outOfService().withDetail("desc", "mock").build());
//    }

}
